package com.ccubee.data.transfer;

import com.alibaba.google.common.util.concurrent.ThreadFactoryBuilder;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.ccubee.data.transfer.event.Cmd;
import com.ccubee.data.transfer.event.EventTypeContext;
import com.ccubee.data.transfer.properties.CanalClientProperties;
import com.ccubee.data.transfer.properties.DataProperties;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * 工作进程
 *
 * @Author ycf
 * @Date 2022/5/22 11:16
 * @Version 1.0
 */
@Slf4j
public class CanalConnectorWork {
    /**
     * canal连接对象
     */
    private CanalConnector canalConnector;
    /**
     * canal客户端配置
     */
    private final CanalClientProperties clientProperties;
    /**
     * 实例名称
     */
    private final String instance;
    /**
     * 表格映射
     */
    private final Map<String, CanalClientProperties.DbMapping> dbMapping;
    /**
     * 线程池
     */
    private final ThreadPoolExecutor executor;
    /**
     * 定时任务
     */
    private final ScheduledExecutorService executorService;
    /**
     * 重试次数
     */
    private int retry = 0;

    /**
     * 默认的构造函数
     *
     * @param dataProperties 配置
     * @param instance       实例名称
     * @param dbMapping      当前实例数据库配置
     */
    public CanalConnectorWork(DataProperties dataProperties, String instance, Map<String, CanalClientProperties.DbMapping> dbMapping) {
        this.instance = instance;
        this.clientProperties = dataProperties.getCanal();
        this.dbMapping = dbMapping;
        DataProperties.InstanceThreadPool instanceThreadPool = dataProperties.getThreadPool(instance);
        executor = new ThreadPoolExecutor(instanceThreadPool.getCorePoolSize(), instanceThreadPool.getMaximumPoolSize(), instanceThreadPool.getKeepAliveTime(), instanceThreadPool.getUnit(), new LinkedBlockingQueue<>(instanceThreadPool.getBlockingQueue()), new ThreadFactoryBuilder()
                .setNameFormat(instance + "-%d").build(), instanceThreadPool.getRejectedType().rejected());
        executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder()
                .setNameFormat(instance + "-%d").build(), instanceThreadPool.getRejectedType().rejected());
        init();
    }

    /**
     * 初始化
     */
    @SneakyThrows
    public void init() {
        connect();
    }

    /**
     * 定时获取数据
     *
     * @param emptyCount 空的次数
     */
    private void loopConnector(int emptyCount) {
        Message message = canalConnector.getWithoutAck(clientProperties.getTcpServer().getBatchSize());
        long batchId = message.getId();
        int size = message.getEntries().size();
        if (batchId == -1 || size == 0) {
            emptyCount++;
        } else {
            emptyCount = 0;
            try {
                printEntry(message.getEntries());
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
        int finalEmptyCount = emptyCount;
        if (log.isDebugEnabled()) {
            log.debug("message emptyCount:{}, batchId:{},size:{}", emptyCount, batchId, size);
        }
        canalConnector.ack(batchId);
        executorService.schedule(() -> loopConnector(finalEmptyCount), 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 连接
     */
    public void connect() {
        canalConnector = CanalConnectors
                .newSingleConnector(
                        new InetSocketAddress(clientProperties.getTcpServer().getHost(), clientProperties.getTcpServer().getPort()),
                        instance, clientProperties.getTcpServer().getUserName(), clientProperties.getTcpServer().getPassword());
        try {
            canalConnector.connect();
            if (clientProperties.getTcpServer().getRetry() > 0 &&
                    retry > clientProperties.getTcpServer().getRetry()) {
                log.error("retry count:{},limit retry count:{}", retry, clientProperties.getTcpServer().getRetry());
                return;
            }
            canalConnector.subscribe(".*\\..*");
            canalConnector.rollback();
            retry = 0;
            log.info("connect server{}, success:{}", instance, clientProperties.getTcpServer());
            loopConnector(0);
        } catch (Exception e) {
            log.error("connect exception:{},retry:{}", e.getMessage(), retry, e);
            disconnect();
            retry++;
            executorService.schedule(this::connect, 2, TimeUnit.SECONDS);
        }

    }

    /**
     * 处理接收到到数据
     *
     * @param entryList 数据集合
     */
    private void printEntry(List<CanalEntry.Entry> entryList) {
        for (CanalEntry.Entry entry : entryList) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                log.error("ERROR ## parser of eromanga-event has an error , data:{}", entry, e);
                continue;
            }
            CanalClientProperties.TableMapping tableMapping;
            if (Objects.isNull(tableMapping = clientProperties.getTableMapping(instance, entry.getHeader().getSchemaName(), entry.getHeader().getTableName()))) {
                if (log.isDebugEnabled()) {
                    log.debug("containsTable not exits,dbName:{},tableName:{}", entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
                }
                continue;
            }
            CanalClientProperties.DbMapping dbMapping = clientProperties.getDbMapping(instance, entry.getHeader().getSchemaName());
            CanalEntry.EventType eventType = rowChange.getEventType();
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                executor.execute(() -> EventTypeContext.cmd(clientProperties.getEventTypeLog(), eventType, Cmd.builder()
                        .entry(entry)
                        .typeProperties(tableMapping)
                        .rowData(rowData)
                        .destination(instance)
                        .dataSource(dbMapping.getDataSource())
                        .schemaName(dbMapping.getDbName(entry.getHeader().getSchemaName()))
                        .tableName(tableMapping.getTableName(entry.getHeader().getTableName())).build()));
            }
        }
    }

    public void disconnect() {
        log.warn("disconnect，destination：{},data:{}", instance, dbMapping);
        canalConnector.disconnect();
        log.info("disconnect success:{}", instance);
    }
}
